package mr.scan;

import java.io.IOException;
import java.util.ArrayList;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import util.LongPair;
import util.Utilties;

public class ScanReducer extends
		Reducer<IntWritable, LongPair, Text, NullWritable> {
	private static final Log LOG = LogFactory.getLog(ScanReducer.class);

	public static String list2String(ArrayList<LongPair> sorted) {
		StringBuffer rst = new StringBuffer(1024);
		for (LongPair pr : sorted) {
			rst.append(pr.first + "=" + pr.second + ";");
		}
		return rst.toString();
	}

	String interSum = "";

	@Override
	protected void cleanup(Context context) throws IOException,
			InterruptedException {
		// context.getConfiguration().set("Scan_test", interSum);
	}

	@Override
	public void reduce(IntWritable key, Iterable<LongPair> values,
			Context context) throws IOException, InterruptedException {
		ArrayList<LongPair> sorted = new ArrayList<LongPair>();

	//	long start = System.currentTimeMillis();
		LOG.info("constructing list");
		for (LongPair val : values) {
			LongPair tmp = new LongPair(val.first, val.second);
			sorted.add(tmp);
		}
	//	LOG.debug("finish constructing list. Time ="
	//			+ (System.currentTimeMillis() - start));

	//	start = System.currentTimeMillis();
	//	LOG.debug("sort list");
		Utilties.sortList(sorted); // sorting, for dis-parallel cases
	//	LOG.debug("sort list. Time =" + (System.currentTimeMillis() - start));

	//	start = System.currentTimeMillis();
		interSum = list2String(sorted);
	//	LOG.debug("to String. Time =" + (System.currentTimeMillis() - start));
		// sorted.
		context.write(new Text(list2String(sorted)), NullWritable.get());

	}
}